package zy.learn.demo.structuredstreaming.window

import java.sql.Timestamp

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
/*
输入和输出：
1,2019-09-14 11:50:00,dog
+---+-------------------+----+
|uid|                 ts|word|
+---+-------------------+----+
|  1|2019-09-14 11:50:00| dog|   watermark为 11:48:00
+---+-------------------+----+
2,2019-09-14 11:51:00,dog
+---+-------------------+----+
|uid|                 ts|word|
+---+-------------------+----+
|  2|2019-09-14 11:51:00| dog|   watermark为 11:49:00
+---+-------------------+----+
1,2019-09-14 11:50:00,cat       没有输出，因为 uid重复，watermark为 11:49:00
3,2019-09-14 11:49:00,dog       没有输出，因为 数据过期，watermark为 11:49:00
3,2019-09-14 11:53:00,dog
+---+-------------------+----+
|uid|                 ts|word|
+---+-------------------+----+
|  3|2019-09-14 11:53:00| dog|   watermark为 11:51:00
+---+-------------------+----+
 */
object StreamDropDuplicate {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().set("spark.sql.shuffle.partitions", "3")
    val spark = SparkSession.builder()
      .master("local[2]")
      .config(sparkConf)
      .appName("StreamDropDuplicate")
      .getOrCreate()

    import spark.implicits._

    val lines = spark.readStream
      .format("socket")
      .option("host", "co7-203")
      .option("port", 9999)
      .load()

    val words = lines.as[String].map(line => {
      val arr: Array[String] = line.split(",")
      (arr(0), Timestamp.valueOf(arr(1)), arr(2))
    }).toDF("uid", "ts", "word")

    val wordCounts = words
      .withWatermark("ts", "2 minutes")
      .dropDuplicates("uid")  // 参数可以传递多个列。uid相同 或者 数据过期，就不再输出

    wordCounts.writeStream
      .outputMode("update")
      .format("console")
      .start
      .awaitTermination()
  }
}
